package com.warren;


import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.source.MySqlSourceBuilder;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @AUTHOR warren
 * @date 2023/12/10
 **/
public class FlinkDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();;

        env.setParallelism(1);

        MySqlSource<String> mySqlSourceBuilder = MySqlSource.<String>builder()
                .hostname("hadoop100")
                .port(3306)
                .databaseList("financial_lease_config")
                .tableList("financial_lease_config.table_process")
                .username("root")
                .password("000000")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();

        DataStreamSource<String> mysqlDBS = env.fromSource(mySqlSourceBuilder, WatermarkStrategy.noWatermarks(), "MysqlFlinkCDC");

        mysqlDBS.print();

        env.execute();
    }
}
